
// KafkaConsumer.cs
using Confluent.Kafka;
using Microsoft.Extensions.Logging;
using System;
using System.Collections.Generic;
using System.Threading;
using System.Threading.Tasks;

namespace Beauty.Net.Kafka
{
    /// <summary>
    /// Kafka 消费者实现
    /// </summary>
    public class KafkaConsumer : IKafkaConsumer
    {
        private readonly IConsumer<string, string> _consumer;
        private readonly ILogger<KafkaConsumer> _logger;
        private readonly KafkaConfig _config;

        public KafkaConsumer(KafkaConfig config, ILogger<KafkaConsumer> logger)
        {
            _config = config;
            _logger = logger;

            var consumerConfig = new ConsumerConfig
            {
                BootstrapServers = config.BootstrapServers,
                GroupId = config.GroupId,
                EnableAutoCommit = config.EnableAutoCommit,
                AutoCommitIntervalMs = config.AutoCommitIntervalMs,
                SessionTimeoutMs = config.SessionTimeoutMs,
                AutoOffsetReset = AutoOffsetReset.Earliest
            };

            // 添加安全配置
            if (!string.IsNullOrEmpty(config.SecurityProtocol))
            {
                if (Enum.TryParse<SecurityProtocol>(config.SecurityProtocol, true, out var securityProtocol))
                {
                    consumerConfig.SecurityProtocol = securityProtocol;
                }
                
                if (!string.IsNullOrEmpty(config.SaslMechanism) && Enum.TryParse<SaslMechanism>(config.SaslMechanism, true, out var saslMechanism))
                {
                    consumerConfig.SaslMechanism = saslMechanism;
                }
                
                consumerConfig.SaslUsername = config.SaslUsername;
                consumerConfig.SaslPassword = config.SaslPassword;
            }

            // 添加额外配置
            foreach (var kvp in config.ConsumerConfig)
            {
                consumerConfig.Set(kvp.Key, kvp.Value);
            }

            _consumer = new ConsumerBuilder<string, string>(consumerConfig).Build();
        }

        /// <summary>
        /// 订阅主题
        /// </summary>
        public void Subscribe(IEnumerable<string> topics)
        {
            _consumer.Subscribe(topics);
            _logger.LogInformation("Subscribed to topics: {Topics}", string.Join(", ", topics));
        }

        /// <summary>
        /// 取消订阅
        /// </summary>
        public void Unsubscribe()
        {
            _consumer.Unsubscribe();
            _logger.LogInformation("Unsubscribed from all topics");
        }

        /// <summary>
        /// 开始消费消息
        /// </summary>
        public async Task StartConsumeAsync(CancellationToken cancellationToken, Func<IKafkaMessage, Task> messageHandler)
        {
            try
            {
                while (!cancellationToken.IsCancellationRequested)
                {
                    var consumeResult = _consumer.Consume(cancellationToken);
                    if (consumeResult is null) continue;

                    var message = new KafkaMessage
                    {
                        Key = consumeResult.Message?.Key ?? string.Empty,
                        Value = consumeResult.Message?.Value ?? string.Empty,
                        Partition = (int)consumeResult.Partition,
                        Timestamp = consumeResult.Message?.Timestamp.UnixTimestampMs
                    };

                    await messageHandler(message);

                    if (!_config.EnableAutoCommit)
                    {
                        _consumer.Commit(consumeResult);
                    }
                }
            }
            catch (OperationCanceledException)
            {
                _logger.LogInformation("Consumer operation was cancelled");
            }
            catch (Exception ex)
            {
                _logger.LogError(ex, "Error occurred while consuming messages");
                throw;
            }
            finally
            {
                _consumer.Close();
            }
        }

        /// <summary>
        /// 手动提交偏移量
        /// </summary>
        public void Commit()
        {
            _consumer.Commit();
        }

        /// <summary>
        /// 异步提交偏移量
        /// </summary>
        public Task CommitAsync()
        {
            _consumer.Commit();
            return Task.CompletedTask;
        }

        public void Dispose()
        {
            _consumer?.Dispose();
        }
    }
}